package org.raymond.iworks.study.mq.kafka.demo;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * Producer如何选择发送的目的分区:
 * 指定分区 partition
 * new ProducerRecord(String topic, Integer partition, ...)
 * 对key的hash值求余得到分区值
 * new ProducerRecord(String topic, K key, V value)
 * 粘性分区,随机选择一个分区,当batch size满或者linger.ms时间到了,才发送消息,之后再随机选择一个分区.
 * 与之前不同,如果随机到与之前的分区相同,则继续随机.
 * new ProducerRecord(String topic, V value)
 *
 * 保证数据不丢的方案:1) 分区副本>=2, 2) acks=-1, 3) min.insync.replicas>=2
 * 生产者造成的数据重复: 以数据不丢的方案为例,数据已经提交副本,leader返回ack之前,leader挂掉.
 * 此时会选举出另一个leader,新leader会重新接收数据.
 *
 * 默认开启幂等,判断重复的条件:PID+Partition+SeqNumber
 *
 * 同一个key,会尽量写入同一个分区,直到分区的缓存满或者linger.ms时间到.这个特性是否可以用来保证消息的顺序?
 *
 * 单分区有序的方法:
 * 如果未开启幂等: max.in.flight.requests.per.connection=1
 * 如果已开启幂等: max.in.flight.requests.per.connection<=5
 * 在kafka1.x以后,启用幂等后,kafka服务器会缓存producer发过来的最近5个request的数据
 */
@Slf4j
public class ProducerTest {
    private static final String kafkaAddress = "192.168.0.125:9094,192.168.0.125:9194,192.168.0.125:9294";

    public static void main(String[] args) {
        produce();
    }

    public static void produce(){
        Properties props = new Properties();
        // 最多配置3台即可,会自动获取元数据
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaAddress);
        // 负责把发送的key字符串转化为字节数组
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 负责把发送的value字符串转化为字节数组
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 设置发送消息的缓冲区,默认32MB
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        // 压缩类型,默认none,不压缩,可以用lz4压缩,会减小数据量,提升吞吐量,但会额外耗费cpu
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
        /**
         * 每当发送多个消息时,为了提高客户端和服务器的性能,生产者将尝试对多个消息进行打包成批,保证这一批可以在同一个分区内.
         * 默认为16384,即16KB.为了尽可能的提高吞吐量,在实际生产中需要对发送的消息进行合理预估,根据实际情况选择一个合理的大小,
         * 避免出现如下情况:
         * 单条消息超过batch.size,Producer有可能不会处理此消息
         * batch.size过大,有可能会造成Producer端内存空间的浪费
         * batch.size过小,频繁的网络IO会降低Producer的吞吐
         */
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 32768);
        /**
         * 如果消息迟迟没有达到batch.size,那么将尝试等待linger.ms时间发送.
         * 默认等待时间为0,也就是当消息到达之后立即发送
         * (但实际上为了减少发送的请求数量,在没有负载的情况下也会延迟5ms的时间.所以这也不是绝对的立即发送)
         * (如果内存不足,也要把消息发出去,以释放内存)
         * 总结:发送条件,batch满或时间到或内存不足,三个条件满足其中一个就会发送
         */
        props.put(ProducerConfig.LINGER_MS_CONFIG, 100);
        /**
         * 配置控制客户端等待请求响应的最长时间.如果在超时时间过去之前未收到响应,则客户端将在必要时重新发送请求,
         * 或者在重试次数用尽时使请求失败
         */
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 10000);
        /**
         * 重试会带来一些问题:
         * 消息会重复: 有的时候一些leader切换之类的问题,需要进行重试,设置retries即可.但是消息重试会导致 重复发送的问题,
         * 比如说网络抖动一下导致他以为没成功,就重试了,其实是成功了
         * 消息乱序:  消息重试是可能导致消息的乱序的,因为可能排在你后面的消息都发送出去了.
         * 所以可以使用"max.in.flight.requests.per.connection"参数设置为1,
         * 这样可以保证producer同一时间只能发送一条消息.
         * 两次重试的间隔默认是100毫秒,用"retry.backoff.ms"来进行设置
         * 基本上在开发过程中,靠重试机制基本就可以搞定95%的异常问题
         */
        // props.put(ProducerConfig.RETRIES_CONFIG, 10); // 5 10
        // props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300);
        /**
         * 建议使用delivery.timeout.ms来控制重试行为,而不是重试次数.默认值2分钟.
         * 调用send()返回后收到成功或失败的时间上限.这限制了消息在发送之前被延迟的总时间、等待broker确认的时间(如果可预期的话)
         * 以及允许可重试发送失败的时间.如果遇到不可恢复的错误、重试次数已用尽、消息被添加到过期批次中,则会发生异常,
         * 此配置的值应大于或等于request.timeout.ms、linger.ms时间的之和
         * 默认开启幂等发送,acks=all,开启幂等发送的情况下不要开重试,因为重试无法去掉重复.
         */
        props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 12000);
        /**
         * producer端设置的 request.required.acks;
         * 0:当消息调用send()发送出去之后就表示消息已经发送成功,不管消息是否已经到达broker
         * 1:消息发送后,leader接收到消息并记录到本地之后,不需要同步数据到副本就能进行ack返回
         * -1:leader接收到消息并记录到本地之后,并且等待副本数据同步完成之后,才会返回ack. 这是默认值
         *
         * ISR:1个副本.1 leader partition, 1 follower partition.
         * kafka服务端:min.insync.replicas:1.如果我们不设置的话,默认这个值是1
         * 一个leader partition会维护一个ISR列表,这个值就是限制ISR列表里面至少有几个副本.
         * 比如这个值是2,那么当ISR列表里面只有一个副本的时候.往这个分区插入数据的时候会报错.
         * 数据不丢失的方案:1) 分区副本>=2, 2) acks=-1, 3) min.insync.replicas>=2
         * 还有可能就是发送有异常:对异常进行处理
         */
        props.put(ProducerConfig.ACKS_CONFIG, "1");

        // 创建一个kafkaProducer,对应一个线程资源,连接各个broker
        Producer<String, String> producer = new KafkaProducer<>(props);
        List<Future<RecordMetadata>> futureList = new ArrayList<>();
        for (int i = 0; i < 3; i++) {
            ProducerRecord record = new ProducerRecord<String, String>("my-topic", Integer.toString(i), Integer.toString(i*2));
            Future<RecordMetadata> futrue = producer.send(record);
            futureList.add(futrue);
        }

        futureList.forEach(f->{
            try {
                RecordMetadata recordMetadata = f.get();
                log.info("{}", recordMetadata.partition());
            } catch (InterruptedException | ExecutionException e) {
                log.error("{}", e.getMessage());
            }
        });
        producer.close();
    }
}
